package cn.linkai.hadoop.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class HBaseDao {

    private static Configuration conf;
    private static Admin admin;
    private static Connection connection;

    /**
     * 初始化hbase连接
     * @throws Exception
     */
    public static void init() throws Exception {
        conf=HBaseConfiguration.create();
//        conf.set("hbase.zookeeper.quorum","192.168.195.140");
//        conf.set("hbase.zookeeper.property.clientPort","2181");
        connection=ConnectionFactory.createConnection();
        admin=connection.getAdmin();
    }

    public static void destroy() throws IOException {
        admin.close();
        connection.close();
    }

    public static void createTable(String tableName,String... columns) throws Exception {
        init();
        if(admin.tableExists(TableName.valueOf(tableName))){
            System.err.println("表已经存在");
        }else{
            HTableDescriptor desc=new HTableDescriptor(TableName.valueOf(tableName));
            for(String columnFamily:columns){
                desc.addFamily(new HColumnDescriptor(columnFamily));
            }
            admin.createTable(desc);
            System.out.println("表创建成功");
        }
    }

    public static void put(String tableName,String row,String columnsFamily,String column,String value) throws IOException {
        HTable table = (HTable) connection.getTable(TableName.valueOf(tableName));
        Put put=new Put(Bytes.toBytes(row));
        put.addColumn(Bytes.toBytes(columnsFamily),Bytes.toBytes(column),Bytes.toBytes(value));
        table.put(put);
    }

    public static void putwithCache(HTable table,String row,String columnsFamily,String column,String value)throws IOException{
        Put put=new Put(Bytes.toBytes(row));
        put.addColumn(Bytes.toBytes(columnsFamily),Bytes.toBytes(column),Bytes.toBytes(value));
        table.put(put);
    }

    public static void main(String[] args) {
        try {
            init();
            createTable("htest","cc");
            long start=System.currentTimeMillis();

//            for(int i=0;i<1000000;i++){
//                put("htest","rk"+i,"cc","test",String.valueOf(i));
//            }
            HTable table= (HTable) connection.getTable(TableName.valueOf("htest"));
            table.setAutoFlushTo(false);
            table.setWriteBufferSize(128*1024);
            for(int i=0;i<1000000;i++){
                putwithCache(table,"rk"+i,"cc","test",String.valueOf(i));
            }
            table.flushCommits();
            long end=System.currentTimeMillis();
            destroy();
            System.out.println("总共花费了:"+(end-start)/1000+" s");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
